package com.itcast.flink.connectors.jdbc;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * <p>Description: </p>
 *
 * @author
 * @version 1.0
 * <p>Copyright:Copyright(c)2020</p>
 * @date
 */
public class MySQLSourceFunction extends RichSourceFunction<AccessLog> {
    /**
     * 数据库连接
     */
    private Connection connection;
    
    /**
     * 预编译SQL
     */
    private PreparedStatement preparedStatement;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 建立Mysql连接
        String jdbcUrl = "jdbc:mysql://192.168.23.140:3306/flink?useSSL=false";
        String executeSql = "select ip, time, type, api from t_access_log";
        connection = DriverManager.getConnection(jdbcUrl, "root", "654321");
        preparedStatement = connection.prepareStatement(executeSql);
    }
    
    @Override
    public void close() throws Exception {
        // 释放数据库连接
        if (null != connection) {
            try {
                connection.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
            connection = null;
        }
    }
    
    @Override
    public void run(SourceContext<AccessLog> ctx) throws Exception {
        // 查询数据库的数据，组装返回
        ResultSet rs = preparedStatement.executeQuery();
        while (rs.next()) {
            AccessLog accessLog = new AccessLog();
            accessLog.setIp(rs.getString("ip"));
            accessLog.setTime(rs.getString("time"));
            accessLog.setType(rs.getString("type"));
            accessLog.setApi(rs.getString("api"));
            ctx.collect(accessLog);
        }
    }
    
    @Override
    public void cancel() {
        System.out.println("cancel method.");
    }
}
